package io.kiki.task.client.common.rpc.client;

import cn.hutool.core.date.StopWatch;
import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.IdUtil;
import io.kiki.task.client.common.annotation.Path;
import io.kiki.task.client.common.cache.GroupVersionCache;
import io.kiki.task.client.common.config.Properties;
import io.kiki.task.client.common.exception.ClientException;
import io.kiki.task.client.common.exception.JobClientTimeOutException;
import io.kiki.task.client.common.exception.JobRemoteException;
import io.kiki.task.common.core.constant.SystemConstants;
import io.kiki.task.common.core.context.SpringContext;
import io.kiki.task.common.core.enums.HeadersEnum;
import io.kiki.task.common.core.enums.StatusEnum;
import io.kiki.task.common.core.model.HttpResponse;
import io.kiki.task.common.core.model.JobRequest;
import io.kiki.task.common.core.model.Result;
import io.kiki.task.common.core.rpc.Future;
import io.kiki.task.common.core.rpc.RpcContext;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.*;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;

/**
 * 请求处理器
 */
@Slf4j
public class RpcClientInvokeHandler<R extends Result<Object>> implements InvocationHandler {

    private static final String HOST_ID = IdUtil.getSnowflake().nextIdStr();
    private static Channel CHANNEL;
    private final Consumer<R> consumer;
    private final boolean async;
    private final long timeout;
    private final TimeUnit timeUnit;

    public RpcClientInvokeHandler(boolean async, long timeout, TimeUnit timeUnit, Consumer<R> consumer) {
        this.consumer = consumer;
        this.async = async;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
    }

    public static void setChannel(Channel channel) {
        RpcClientInvokeHandler.CHANNEL = channel;
    }

    /**
     * 发送数据
     *
     * @param url  url地址
     * @param body 请求的消息体
     * @throws InterruptedException
     */
    public static void send(String url, String body) throws InterruptedException {

        if (Objects.isNull(RpcClientInvokeHandler.CHANNEL)) {
            log.error("send message but channel is null url:[{}] method:[{}] body:[{}] ", url, body);
            return;
        }

        // 配置HttpRequest的请求数据和一些配置信息
        FullHttpRequest fullHttpRequest = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, url, Unpooled.wrappedBuffer(body.getBytes(StandardCharsets.UTF_8)));
        Properties properties = SpringContext.getBean(Properties.class);

        // server配置不能为空
        Properties.ServerConfig serverConfig = properties.getServerConfig();
        if (Objects.isNull(serverConfig)) {
            log.error(" job server config is null");
            return;
        }

        Assert.notBlank(properties.getGroup(),
                () -> new JobRemoteException("The group is null, please check if your configuration is correct."));

        fullHttpRequest.headers()
                .set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON)
                //开启长连接
                .set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE)
                //设置传递请求内容的长度
                .set(HttpHeaderNames.CONTENT_LENGTH, fullHttpRequest.content().readableBytes())
                .set(HeadersEnum.HOST_ID.getKey(), HOST_ID)
                .set(HeadersEnum.HOST_IP.getKey(), NettyChannel.getClientHost())
                .set(HeadersEnum.GROUP_NAME.getKey(), properties.getGroup())
                .set(HeadersEnum.HOST_PORT.getKey(), NettyChannel.getClientPort())
                .set(HeadersEnum.VERSION.getKey(), GroupVersionCache.getVersion())
                .set(HeadersEnum.HOST.getKey(), serverConfig.getHost())
                .set(HeadersEnum.NAMESPACE.getKey(), Optional.ofNullable(properties.getNamespace()).orElse(
                        SystemConstants.DEFAULT_NAMESPACE))
                .set(HeadersEnum.TOKEN.getKey(), Optional.ofNullable(properties.getToken()).orElse(
                        SystemConstants.DEFAULT_TOKEN))
        ;

        //发送数据
        try {
            CHANNEL.writeAndFlush(fullHttpRequest).sync();
        } catch (Exception exception) {
            throw new JobRemoteException("网络异常");
        }
    }

    @Override
    public R invoke(final Object proxy, final Method method, final Object[] args) throws Throwable {

        Path path = method.getAnnotation(Path.class);
        JobRequest jobRequest = new JobRequest(args);
        Future<R> future = Future.newFuture(jobRequest.getReqId(), timeout, timeUnit);
        RpcContext.setFuture(future);
        StopWatch stopWatch = new StopWatch();
        stopWatch.start("request start " + jobRequest.getReqId());
        try {
            send(path.value(), jobRequest.toString());
        } finally {
            stopWatch.stop();
        }
        log.debug("request complete requestId:[{}] 耗时:[{}ms]", jobRequest.getReqId(), stopWatch.getTotalTimeMillis());
        if (async) {
            future.whenComplete((r, t) -> {
                if (Objects.nonNull(t)) {
                    consumer.accept(
                            (R) new HttpResponse(StatusEnum.NO.getStatus(), t.getMessage(), null, jobRequest.getReqId()));
                } else {
                    consumer.accept(r);
                }
            });
            return null;
        } else {
            Assert.notNull(future, () -> new ClientException("completableFuture is null"));
            try {
                return future.get(Integer.MAX_VALUE, TimeUnit.MILLISECONDS);
            } catch (ExecutionException e) {
                throw e.getCause();
            } catch (TimeoutException e) {
                throw new JobClientTimeOutException("Request to remote interface timed out. path:[{}]", path.value());
            }
        }

    }

}
